package com.markhsiu.MiniMQ.remote.transport.netty;

import java.util.concurrent.TimeUnit;

import com.markhsiu.MiniMQ.broker.message.BrokerFactory;
import com.markhsiu.MiniMQ.model.Message;
import com.markhsiu.MiniMQ.model.Result;
import com.markhsiu.MiniMQ.remote.Client;
import com.markhsiu.MiniMQ.remote.URL;
import com.markhsiu.MiniMQ.remote.transport.netty.handler.NettyClientHandler;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

/**
 * Created by Mark Hsiu on 2017/2/8.
 */
public class NettyClient implements Client{

    private EventLoopGroup worker = new NioEventLoopGroup();
    private Bootstrap bootstrap;

    private URL url ;
    private Channel messageChannel;
    public  NettyClient(){
        url = new URL();
    }

    public  NettyClient(URL serverURL){
        if(serverURL == null){
            throw new RuntimeException(" server URL is null");
        }
        url = serverURL;
    }

    @Override
    public Result send(Message message){
        final Result result = new Result();
        
        BrokerFactory.instance().putCallBack(message.getMessageID(),result);
        ChannelFuture channel = messageChannel.writeAndFlush(message);
        channel.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    System.out.println(future.cause());
                    result.setError(1);
                }
            }
        });

        Message ask = result.getMessageResult(10, TimeUnit.SECONDS);
        if(ask == null){
            if(result.getError() == 1){
                throw new RuntimeException("响应超时");
            } else {
                System.out.println("结果为null");
            }
        }

        result.setMessage(ask);
        return result;
    }

    @Override
    public URL getServerURL() {
        return url;
    }

    @Override
    public void open(){
        bootstrap = new Bootstrap();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .handler(new ChilderChannelHandler());

    }

    @Override
    public void connect() {
        if(bootstrap == null){
            throw new RuntimeException(" you need open Client before connect");
        }

        try {
            ChannelFuture  future = bootstrap.connect(url.getHostname(),url.getPort()).sync();
            future.addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture future) throws Exception {
                    Channel channel = future.channel();
                    messageChannel = channel;
                }
            });

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() {
        try {
            if(messageChannel != null){
                messageChannel.close().sync();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            worker.shutdownGracefully();
        }

    }

    private  class ChilderChannelHandler extends ChannelInitializer<SocketChannel>{

        @Override
        protected void initChannel(SocketChannel channel) throws Exception {
            ChannelPipeline pip = channel.pipeline();
            // 1. 禁止对类加载器进行缓存，它在基于OSGI的动态模块化编程中经常使用
            pip.addLast(new ObjectDecoder(1024,
                    ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
            pip.addLast(new ObjectEncoder());

            pip.addLast(new NettyClientHandler());
        }
    }

}
